
#include "RtErrorNetwork.h"
#include "RtThreadManager.h"
#include "RtUtilTemplates.h"
#include "CmDBThread.h"
#include "RtTrace.h"

#define MAX_CONTENT_LEN	4000

#ifdef __ORACLE__
///////////////////////////////////////////
//class CDBStoredProcedureHelper
///////////////////////////////////////////
CDBStoredProcedureHelper::CDBStoredProcedureHelper()
{
	m_pDB = NULL;
}

CDBStoredProcedureHelper::~CDBStoredProcedureHelper()
{
	Clean();
}

void CDBStoredProcedureHelper::Clean()
{
	char* pTmp = NULL;
	while(!m_vstrInTmps.empty())
	{
		pTmp = m_vstrInTmps.back();
		m_vstrInTmps.pop_back();
		delete[] pTmp;
	}

	DWORD* pdwTmp = NULL;
	while(!m_vIntInTmps.empty())
	{
		pdwTmp = m_vIntInTmps.back();
		m_vIntInTmps.pop_back();
		delete pdwTmp;
	}

	std::map<CRtString, char*>::iterator it_stop1 = m_mOutStrParams.begin();

	for(; it_stop1 != m_mOutStrParams.end(); it_stop1++)
	{
		delete[] (it_stop1->second);
	}
	m_mOutStrParams.clear();

	std::map<CRtString, DWORD*>::iterator it_stop2 = m_mOutIntParams.begin();

	for(; it_stop2 != m_mOutIntParams.end(); it_stop2++)
	{
		delete (it_stop2->second);
	}
	m_mOutIntParams.clear();

	m_pDB = NULL;
}

RtResult CDBStoredProcedureHelper::
PrepareProcedure(CRtString& strProcedure, CDBOciAdaptor* pDB)
{
	RT_ASSERTE(pDB);

	Clean();	// Clean contents of last execution

	m_pDB = pDB;

	CRtString strSQL = "BEGIN ";
	strSQL += strProcedure;
	strSQL += "; ";
	strSQL += "END;";

	return m_pDB->PrepareStmt(strSQL.c_str());
}

RtResult CDBStoredProcedureHelper::
BindParam(
		  LPCSTR szParameter, 
		  DWORD dwParam, ParamType pType)
{
	RT_ASSERTE(m_pDB);

	RtResult rv;

	CRtString strParameter = szParameter;

	DWORD* pdwTmp = new DWORD;
	switch(pType) 
	{
	case PARAM_TYPE_IN:
		{
			*pdwTmp = dwParam;
			rv = m_pDB->BindStmtInParam(szParameter, *pdwTmp);
			m_vIntInTmps.push_back(pdwTmp);
		}
		break;
	case PARAM_TYPE_OUT:
		{
			*pdwTmp = 0;
			rv = m_pDB->BindStmtOutParam(szParameter, *pdwTmp);
			m_mOutIntParams[strParameter] = pdwTmp;
		}
		break;
	case PARAM_TYPE_IN_OUT:
		{
			*pdwTmp = dwParam;
			rv = m_pDB->BindStmtOutParam(szParameter, *pdwTmp);
			m_mOutIntParams[strParameter] = pdwTmp;
		}
		break;
	default:
		RT_ERROR_TRACE("CDBStoredProcedureHelper::BindParam(), unknow ParamType");
	}

	return rv;
}

RtResult CDBStoredProcedureHelper::
BindParam(
		  LPCSTR szParameter, 
		  LPCSTR szParam, ParamType pType)
{
	RT_ASSERTE(m_pDB);

	RtResult rv;

	CRtString strParameter = szParameter;

	int nParamLen = strlen(szParam);

	if(pType == PARAM_TYPE_IN)
		RT_ASSERTE(nParamLen >= 0);

	RT_ASSERTE(nParamLen < MAX_CONTENT_LEN);

	char* pTmp = NULL;

	switch(pType) 
	{
	case PARAM_TYPE_IN:
		{
			pTmp = new char[nParamLen + 1];
			memset(pTmp, 0, nParamLen + 1);
			strcpy(pTmp, szParam);

			rv = m_pDB->BindStmtInParam(szParameter, (LPBYTE)pTmp);

			m_vstrInTmps.push_back(pTmp);
		}
		break;
	case PARAM_TYPE_OUT:
		{
			pTmp = new char[MAX_CONTENT_LEN];
			memset(pTmp, 0, MAX_CONTENT_LEN);
			rv = m_pDB->BindStmtOutParam(szParameter, (LPBYTE)pTmp);

			m_mOutStrParams[strParameter] = pTmp;
		}
		break;
	case PARAM_TYPE_IN_OUT:
		{
			pTmp = new char[MAX_CONTENT_LEN];
			memset(pTmp, 0, MAX_CONTENT_LEN);
			strcpy(pTmp, szParam);
			rv = m_pDB->BindStmtOutParam(szParameter, (LPBYTE)pTmp);

			m_mOutStrParams[strParameter] = pTmp;
		}
		break;
	default:
		RT_ERROR_TRACE("CDBStoredProcedureHelper::BindParam(), unknow ParamType");
	}

	return rv;
}

RtResult CDBStoredProcedureHelper::ExcuteProcedure()
{
	RT_ASSERTE(m_pDB);

	RtResult rv;
	rv = m_pDB->ExecutePreparedSQLStmt();

	char* pTmp = NULL;
	while(!m_vstrInTmps.empty())
	{
		pTmp = m_vstrInTmps.back();
		m_vstrInTmps.pop_back();
		delete[] pTmp;
	}

	DWORD* pdwTmp = NULL;
	while(!m_vIntInTmps.empty())
	{
		pdwTmp = m_vIntInTmps.back();
		m_vIntInTmps.pop_back();
		delete pdwTmp;
	}

	return rv;
}

RtResult CDBStoredProcedureHelper::
GetParamValue(LPCSTR szParameter, DWORD &dwVal)
{
	CRtString strParameter = szParameter;
	std::map<CRtString, DWORD*>::iterator it_stop = m_mOutIntParams.begin();

	for(; it_stop != m_mOutIntParams.end(); it_stop++)
	{
		if(it_stop->first == strParameter)
		{
			dwVal = *(it_stop->second);
			break;
		}
	}

	if(it_stop == m_mOutIntParams.end()) // Cannot find
		return RT_ERROR_FAILURE;

	return RT_OK;
}
	
RtResult CDBStoredProcedureHelper::
GetParamValue(LPCSTR szParameter, CRtString &strVal)
{
	CRtString strParameter = szParameter;
	std::map<CRtString, char*>::iterator it_stop = m_mOutStrParams.begin();

	for(; it_stop != m_mOutStrParams.end(); it_stop++)
	{
		if(it_stop->first == strParameter)
		{
			strVal = it_stop->second;
			break;
		}
	}

	if(it_stop == m_mOutStrParams.end()) // Cannot find
		return RT_ERROR_FAILURE;

	return RT_OK;
}

///////////////////////////////////////////
//class CDBParaBindSQLHelper
///////////////////////////////////////////
CDBParaBindSQLHelper::CDBParaBindSQLHelper()
{
	m_pDB = NULL;
	m_bGetRecordSet = FALSE;
}

CDBParaBindSQLHelper::~CDBParaBindSQLHelper()
{
	Clean();
}

void CDBParaBindSQLHelper::Clean()
{
	char* pTmp = NULL;
	while(!m_vstrInTmps.empty())
	{
		pTmp = m_vstrInTmps.back();
		m_vstrInTmps.pop_back();
		delete[] pTmp;
	}

	DWORD* pdwTmp = NULL;
	while(!m_vIntInTmps.empty())
	{
		pdwTmp = m_vIntInTmps.back();
		m_vIntInTmps.pop_back();
		delete pdwTmp;
	}
	m_pDB = NULL;

	m_bGetRecordSet = FALSE;
}

RtResult CDBParaBindSQLHelper::
StrReplace(CRtString& strSQL, CRtString strParam, CRtString strVal)
{
	int n = strSQL.find(strParam);
	if(n == -1)// Can not find
		return RT_ERROR_FAILURE;

	strSQL.replace(n, strParam.length(), strVal);
	return RT_OK;
}

RtResult CDBParaBindSQLHelper::
PrepareSQL(CRtString& strSQL, CDBOciAdaptor* pDB, BOOL bGetRecordSet)
{
	RT_ASSERTE(pDB);

	Clean();	// Clean contents of last execution

	m_pDB = pDB;
	m_bGetRecordSet = bGetRecordSet;

	if(m_bGetRecordSet)
	{
		m_strSQLSelect = strSQL;
		return RT_OK;
	}

	return m_pDB->PrepareStmt(strSQL.c_str());
}

RtResult CDBParaBindSQLHelper::
BindParam(LPCSTR szParameter, DWORD dwParam)
{
	RT_ASSERTE(m_pDB);

	RtResult rv;

	if(m_bGetRecordSet)
	{
		CRtString strVal = CDBHelper::Number2String(dwParam);
		rv = StrReplace(m_strSQLSelect, szParameter, strVal);
	}
	else
	{
		DWORD* pdwTmp = new DWORD;
		
		*pdwTmp = dwParam;
		rv = m_pDB->BindStmtInParam(szParameter, *pdwTmp);
		m_vIntInTmps.push_back(pdwTmp);
	}
	
	return rv;
}

RtResult CDBParaBindSQLHelper::
BindParam(LPCSTR szParameter, LPCSTR szParam)
{
	RT_ASSERTE(m_pDB);

	RtResult rv;

	if(m_bGetRecordSet)
	{
		CRtString strVal = "'";
		strVal += szParam;
		strVal += "'";
		rv = StrReplace(m_strSQLSelect, szParameter, strVal);
	}
	else
	{
		int nParamLen = strlen(szParam);
		
		RT_ASSERTE(nParamLen >= 0 && nParamLen < MAX_CONTENT_LEN);
		
		char* pTmp = NULL;
		
		pTmp = new char[nParamLen + 1];
		memset(pTmp, 0, nParamLen + 1);
		strcpy(pTmp, szParam);
		
		rv = m_pDB->BindStmtInParam(szParameter, (LPBYTE)pTmp);
		
		m_vstrInTmps.push_back(pTmp);
	}

	return rv;
}
	
RtResult CDBParaBindSQLHelper::ExcuteSQL()
{
	RT_ASSERTE(m_pDB);

	RtResult rv;

	if(m_bGetRecordSet)
	{
		rv = DoSQL(m_strSQLSelect, m_pDB, TRUE);
	}
	else
	{
		rv = m_pDB->ExecutePreparedSQLStmt();
	}

	Clean();

	return rv;
}
#endif

DWORD CDBHelper::g_dwExNum = 0;
///////////////////////////////////////////
//class CDBHelper
///////////////////////////////////////////
CDBHelper::CDBHelper()
{
	m_pDB = NULL;

	m_pRecordSet = NULL;
	m_wCols = 0;
	m_wRows = 0;
	m_wRowIndex = 0;
}

CDBHelper::~CDBHelper()
{
}

RtResult CDBHelper::DoSQL(const CRtString &strSQL, CDBOciAdaptor* pDB, BOOL bGetRecordSet)
{
	RT_ASSERTE(pDB);
	m_pDB = pDB;

	RtResult rv = RT_OK;

	g_dwExNum++;
	m_pRecordSet = NULL;
	m_wCols = 0;
	m_wRows = 0;
	m_wRowIndex = 0;
	m_wAffectRows = 0;
	
	if(bGetRecordSet)
	{
		rv = pDB->GetDataFromDB(strSQL.c_str());
		
		if(RT_FAILED(rv))
		{
			return rv;
		}
		
		m_pRecordSet = pDB->GetRecordSet();
		RT_ASSERTE(m_pRecordSet);
		m_wRows = m_pRecordSet->GetRecordCount();
		m_wCols = m_pRecordSet->GetColumnCount();
		m_wAffectRows = m_pRecordSet->GetAffectRows();
	}
	else
	{
		rv = pDB->PutDataInDB(strSQL.c_str());
		
		if(RT_FAILED(rv))
		{
//			RT_ERROR_TRACE("CDBHelper::DoSQL(), PutDataInDB failed.");
			return rv;
		}
	}

	return rv;
}

RtResult CDBHelper::DoSQLAutoInc(const CRtString& strSQL, CDBOciAdaptor* pDB, LONGLONG& ulAutoValue)
{
	RT_ASSERTE(pDB);
	m_pDB = pDB;

	RtResult rv = RT_OK;

	m_pRecordSet = NULL;
	m_wCols = 0;
	m_wRows = 0;
	m_wRowIndex = 0;
	
	rv = pDB->InsertAutoIncRow(strSQL.c_str(), ulAutoValue);
		
	if(RT_FAILED(rv))
	{
//			RT_ERROR_TRACE("CDBHelper::DoSQL(), PutDataInDB failed.");
		return rv;
	}

	return rv;
}

RtResult CDBHelper::DoProcedure(CRtString &strProcedure, CDBOciAdaptor* pDB)
{
	RtResult rv = RT_OK;
#ifdef __ORACLE__
	RT_ASSERTE(pDB);
	m_pDB = pDB;
	
	rv = pDB->PutDataInDB(strProcedure.c_str(), IS_PROCEDURE);
	
	if(RT_FAILED(rv))
	{
		RT_ERROR_TRACE("CDBHelper::DoProcedure(), execution failed.");
		return rv;
	}
#endif	
	return rv;
}

WORD CDBHelper::GetRecordRows()
{
	//RT_ASSERTE(m_pRecordSet);

	return m_wRows;
}

WORD CDBHelper::GetRecordCols()
{
	//RT_ASSERTE(m_pRecordSet);

	return m_wCols;
}

void CDBHelper::GetOCIValue(WORD wFieldIndex, void* &pvVal)
{
	RT_ASSERTE(m_pRecordSet);
	RT_ASSERTE(m_wRowIndex < m_wRows);// m_wRowIndex = [0, m_wRows)
	RT_ASSERTE(wFieldIndex < m_wCols);// wFieldIndex = [0, m_wCols)
#ifdef __ORACLE__
	CDBRecord_* pRecord = NULL;
	CVariant_* pVariant = NULL;
	//For temp use
	LPBYTE pData;
	DWORD dwSize;
	short nType;

	pRecord = m_pRecordSet->GetRecord(m_wRowIndex);
	RT_ASSERTE(pRecord);
	
	//Get field Value
	pVariant = pRecord->GetDataItem(wFieldIndex);
	RT_ASSERTE(pVariant);
	pVariant->GetValue(&nType, &pData, &dwSize);

	if(nType == SQLT_AFC)	//For : Ansi fixed char [ Table Field like CHAR(1)]
	{
		m_ch[0] = *(char*)pData;
		m_ch[1] = '\0';

		pvVal = (void*)m_ch;
	}
	else
		pvVal = pData;
#else
	MYSQL_ROW row = m_pRecordSet->GetRecord(m_wRowIndex);
	RT_ASSERTE(row);
	pvVal = (void*)row[wFieldIndex];
#endif
}

RtResult CDBHelper::GetString(WORD wFieldIndex, CRtString &strVal)
{
	void* pvVal = NULL;
	GetOCIValue(wFieldIndex, pvVal);

	if(pvVal == NULL)
		strVal = "";
	else
		strVal = (char*)pvVal;

	return RT_OK;
}

RtResult CDBHelper::GetShort(WORD wFieldIndex, SHORT &sVal)
{
	void* pvVal = NULL;
	GetOCIValue(wFieldIndex, pvVal);
	RtResult rv = RT_OK;
#ifdef __ORACLE__
	WORD wTmp = 0;
	rv = m_pDB->GetIntFromOCINumber((OCINumber*)pvVal, wTmp);
	sVal = wTmp;
#else
	if(pvVal)
		sVal = atoi((const char*)pvVal);
	else
		sVal = 0;
#endif
	return rv;
}

RtResult CDBHelper::GetUShort(WORD wFieldIndex, WORD &usVal)
{
	void* pvVal = NULL;
	GetOCIValue(wFieldIndex, pvVal);
	RtResult rv = RT_OK;
#ifdef __ORACLE__
	DWORD dwTmp = 0;
	rv = m_pDB->GetUIntFromOCINumber((OCINumber*)pvVal, dwTmp);
	usVal = dwTmp;
#else
	if(pvVal)
		usVal = (WORD)atoi((const char*)pvVal);
	else
		usVal = 0;
#endif

	return rv;
}

RtResult CDBHelper::GetLong(WORD wFieldIndex, LONG &lVal)
{
	void* pvVal = NULL;
	GetOCIValue(wFieldIndex, pvVal);
	RtResult rv = RT_OK;
#ifdef __ORACLE__
	DWORD dwTmp = 0;
	RtResult rv = m_pDB->GetIntFromOCINumber((OCINumber*)pvVal, dwTmp);
	lVal = dwTmp;
#else
	if(pvVal)
		lVal = atol((const char*)pvVal);
	else
		lVal = 0;
#endif

	return rv;
}

RtResult CDBHelper::GetULong(WORD wFieldIndex, DWORD &ulVal)
{
	void* pvVal = NULL;
	GetOCIValue(wFieldIndex, pvVal);
	RtResult rv = RT_OK;
#ifdef __ORACLE__
	DWORD dwTmp = 0;
	RtResult rv = m_pDB->GetUIntFromOCINumber((OCINumber*)pvVal, dwTmp);
	ulVal = dwTmp;
#else
	if(pvVal)
		ulVal = (DWORD)atol((const char*)pvVal);
	else
		ulVal = 0;
#endif

	return rv;
}

RtResult CDBHelper::GetLongLong(WORD wFieldIndex, LONGLONG &llVal)
{
	void* pvVal = NULL;
	GetOCIValue(wFieldIndex, pvVal);
	RtResult rv = RT_OK;
#ifdef __ORACLE__

#error Unimplemented yet!

#else
	if(pvVal)
#ifdef RT_WIN32
		llVal = _atoi64((const char*)pvVal);
#else
		llVal = atoll((const char*)pvVal);
#endif
	else
		llVal = 0;
#endif

	return rv;
}

RtResult CDBHelper::GetReal(WORD wFieldIndex, double &rVal)
{
	void* pvVal = NULL;
	GetOCIValue(wFieldIndex, pvVal);
	RtResult rv = RT_OK;
#ifdef __ORACLE__
	return m_pDB->GetRealFromOCINumber((OCINumber*)pvVal, rVal);
#else
	if(pvVal)
		rVal = atof((const char*)pvVal);
	else
		rVal = 0;
#endif
	return rv;
}

#ifdef __ORACLE__
RtResult CDBHelper::GetGMTTime(WORD wFieldIndex, gmt_time &tm)
{
	void* pvVal = NULL;
	GetOCIValue(wFieldIndex, pvVal);
	return m_pDB->GetGMTTimeFromOCIDate((OCIDate*)pvVal, &tm);
}
#endif

RtResult CDBHelper::GetNextRow()
{
	if((m_wRowIndex + 1) == m_wRows)// Last row
	{
		return RT_ERROR_NOT_AVAILABLE;
	}

	++m_wRowIndex;

	return RT_OK;
}

char* CDBHelper::Number2String(DWORD dwNumber)
{
	static char m_pNum2StrResult[20];
	memset(m_pNum2StrResult, 0, 20);
	
	sprintf(m_pNum2StrResult, "%u", dwNumber);
	return m_pNum2StrResult;
}

DWORD CDBHelper::String2Number(const char* str)
{
	int nLen = strlen(str);
	DWORD dwRet = 0;
	int j = 1;	//For power 10
	for(int i=nLen-1; i>=0; i--)
	{
		dwRet += (str[i] - 0x30) * j;
		j *= 10;
	}
	return dwRet;
}

char* CDBHelper::GetCurrentDate()
{
	static char m_pCurrData[20];
	
	time_t curTime;
	struct tm *curTimeDecoded;
	
	time(&curTime);
	curTimeDecoded = localtime(&curTime);
	
	int nSec = curTimeDecoded->tm_sec;
	int nMin = curTimeDecoded->tm_min;
	int nHour = curTimeDecoded->tm_hour;
	int nDay = curTimeDecoded->tm_mday;
	int nMon = curTimeDecoded->tm_mon + 1;
	int nYear = curTimeDecoded->tm_year + 1900;
	
	memset(m_pCurrData ,0, 20);
	//dd-mm-yy HH24:MI:SS
	sprintf(m_pCurrData, "%02d-%02d-%04d %02d:%02d:%02d", nDay, nMon, nYear, nHour, nMin, nSec);
	
	return m_pCurrData;
}
///////////////////////////////////////////
//class CDBThreadManager
///////////////////////////////////////////
// CDBThreadManager* CDBThreadManager::Instance()
// {
// 	return CRtSingletonT<CDBThreadManager>::Instance();
// }

CDBThreadManager::CDBThreadManager()
{
	m_pCurrEventQueue 
		= CRtThreadManager::Instance()->GetThreadEventQueue(CRtThreadManager::TT_CURRENT);

	m_wThreadNum = 0;
	m_wCurrentIndex = 0;
	m_nPushNumber = 0;
	m_wDBType = DB_TYPE_MYSQL;
}

CDBThreadManager::~CDBThreadManager()
{
	RT_INFO_TRACE("CDBThreadManager::~CDBThreadManager");
	Clean();
}	

RtResult CDBThreadManager::Initialize(LPCSTR szDBSvr,
									  LPCSTR szDBName,
									  LPCSTR szUserName,
									  LPCSTR szPasswd,
									  WORD wThreadNum,
									  BOOL bThreadAutoSelect,
									  WORD wDBType)
{
	/*
	RtResult rv = RT_OK;
	RtResult rvDB = RT_DB_SUCCESS;
	CRtThread* pDBThread = NULL;
	HANDLE pDB = NULL;
	m_wDBType = wDBType;
	
	for(WORD wI = 0; wI < wThreadNum; wI++)
	{
		rv = CRtThreadManager::Instance()->CreateUserTaskThread(pDBThread);
		
		if(RT_FAILED(rv))
		{
			RT_ERROR_TRACE("CDBThreadManager::Initialize(), create No. " << (wI + 1) << " DB Thread failed");
			Clean();
			return rv;
		}
		
		m_vDBThreads.push_back(pDBThread);
		
		switch(wDBType)
		{
		case DB_TYPE_MYSQL:
			{
				CDBOciAdaptor* pAdaptor = new CDBOciAdaptor(szDBSvr, szDBName, szUserName, szPasswd);
				rvDB = pAdaptor->OpenDatabase();				
				if(RT_FAILED(rvDB))
				{
					RT_ERROR_TRACE("CDBThreadManager::Initialize(), Connect to DB for No. " << (wI + 1) << " DB Thread failed");
					pAdaptor->CloseDatabase();
					delete pAdaptor;					
					Clean();
					return RT_ERROR_FAILURE;
				}
				pDB = (HANDLE)pAdaptor;
			}
			break;
		case DB_TYPE_PGSQL:
			{
				_pg pg_param;
				pg_param.index = 0;
				pg_param.port = 1921;//5432;
				memcpy(pg_param.ipaddr, szDBSvr, strlen(szDBSvr) + 1);
				memcpy(pg_param.dbname, szDBName, strlen(szDBName) + 1);
				memcpy(pg_param.dbuser, szUserName, strlen(szUserName)+1);
				memcpy(pg_param.dbpwd, szPasswd, strlen(szPasswd)+1);
				pDB = SkyRC_OpenDB(rs_pg, (char*)&pg_param, sizeof(_pg));
			}
			break;
		case DB_TYPE_MEMCACHE:
			{
				_mc mc_param;
				memcpy(mc_param.ipaddr, szDBSvr, strlen(szDBSvr)+1);
				mc_param.port = 11211;
				pDB = SkyRC_OpenDB(rs_mc, (char*)&mc_param, sizeof(_mc));
			}
			break;
		case DB_TYPE_MGDB:
			{
				_mg mg_param;
				mg_param.port = 5281;//11411;//27017;
				memcpy(mg_param.ipaddr, szDBSvr, strlen(szDBSvr)+1);
				memcpy(mg_param.dbname, szDBName, strlen(szDBName)+1);
				memcpy(mg_param.dbuser, szUserName, strlen(szUserName)+1);
				memcpy(mg_param.dbpwd, szPasswd, strlen(szPasswd)+1);
				pDB = SkyRC_OpenDB(rs_mg, (char*)&mg_param, sizeof(_mg));
			}
			break;
		}				
		if(NULL == pDB)
		{
			RT_ERROR_TRACE("CDBThreadManager::Initialize(), Connect to DB for No. " << (wI + 1) << " DB Thread failed");					
			Clean();
			return RT_ERROR_FAILURE;
		}
		
		m_vDBAdaptors.push_back(pDB);

		pDBThread = NULL;
		pDB = NULL;
	}

	m_wThreadNum = wThreadNum;
	m_bThreadAutoSelect = bThreadAutoSelect;

	//RT_INFO_TRACE("CDBThreadManager::Initialize(), create " << wThreadNum << " DB Threads");
	return rv;
	*/
	
	unsigned short nPort = 0;
	switch(wDBType)
	{
	case DB_TYPE_MYSQL:
		{
		}
		break;
	case DB_TYPE_PGSQL:
		{
			nPort = 1921;//5432;
		}
		break;
	case DB_TYPE_MEMCACHE:
		{
			nPort = 11213;//11211;
		}
		break;
	case DB_TYPE_MGDB:
		{
			nPort = 5281;//11411;//27017;
		}
		break;
	}
	return Initialize2(szDBSvr, nPort, szDBName, szUserName, szPasswd, wThreadNum, bThreadAutoSelect, wDBType);
}

RtResult CDBThreadManager::Initialize2(LPCSTR szDBSvr,
									  unsigned short nPort,
									  LPCSTR szDBName,
									  LPCSTR szUserName,
									  LPCSTR szPasswd,
									  WORD wThreadNum,
									  BOOL bThreadAutoSelect,
									  WORD wDBType)
{
	RT_INFO_TRACE("CDBThreadManager::Initialize2, db:"<<szDBSvr<<":"<<nPort<<", dbname:"<<szDBName<<", user:"<<szUserName<<", pwd:"<<szPasswd);

	RtResult rv = RT_OK;
	RtResult rvDB = RT_DB_SUCCESS;
	CRtThread* pDBThread = NULL;
	HANDLE pDB = NULL;
	m_wDBType = wDBType;
	
	for(WORD wI = 0; wI < wThreadNum; wI++)
	{
		rv = CRtThreadManager::Instance()->CreateUserTaskThread(pDBThread);
		
		if(RT_FAILED(rv))
		{
			RT_ERROR_TRACE("CDBThreadManager::Initialize(), create No. " << (wI + 1) << " DB Thread failed");
			Clean();
			return rv;
		}
		
		m_vDBThreads.push_back(pDBThread);
		
		switch(wDBType)
		{
		case DB_TYPE_MYSQL:
			{
				CDBOciAdaptor* pAdaptor = new CDBOciAdaptor(szDBSvr, szDBName, szUserName, szPasswd);
				rvDB = pAdaptor->OpenDatabase();				
				if(RT_FAILED(rvDB))
				{
					RT_ERROR_TRACE("CDBThreadManager::Initialize(), Connect to DB for No. " << (wI + 1) << " DB Thread failed");
					pAdaptor->CloseDatabase();
					delete pAdaptor;					
					Clean();
					return RT_ERROR_FAILURE;
				}
				pDB = (HANDLE)pAdaptor;
			}
			break;
		case DB_TYPE_PGSQL:
			{
				_pg pg_param;
				pg_param.index = 0;
				pg_param.port = nPort;
				memcpy(pg_param.ipaddr, szDBSvr, strlen(szDBSvr) + 1);
				memcpy(pg_param.dbname, szDBName, strlen(szDBName) + 1);
				memcpy(pg_param.dbuser, szUserName, strlen(szUserName)+1);
				memcpy(pg_param.dbpwd, szPasswd, strlen(szPasswd)+1);
				pDB = SkyRC_OpenDB(rs_pg, (char*)&pg_param, sizeof(_pg));
			}
			break;
		case DB_TYPE_MEMCACHE:
			{
				_mc mc_param;
				memcpy(mc_param.ipaddr, szDBSvr, strlen(szDBSvr)+1);
				mc_param.port = nPort;
				pDB = SkyRC_OpenDB(rs_mc, (char*)&mc_param, sizeof(_mc));
			}
			break;
		case DB_TYPE_MGDB:
			{
				_mg mg_param;
				mg_param.port = nPort;
				memcpy(mg_param.ipaddr, szDBSvr, strlen(szDBSvr)+1);
				memcpy(mg_param.dbname, szDBName, strlen(szDBName)+1);
				memcpy(mg_param.dbuser, szUserName, strlen(szUserName)+1);
				memcpy(mg_param.dbpwd, szPasswd, strlen(szPasswd)+1);
				pDB = SkyRC_OpenDB(rs_mg, (char*)&mg_param, sizeof(_mg));
			}
			break;
		}				
		if(NULL == pDB)
		{
			RT_ERROR_TRACE("CDBThreadManager::Initialize(), Connect to DB for No. " << (wI + 1) << " DB Thread failed");					
			Clean();
			return RT_ERROR_FAILURE;
		}
		
		m_vDBAdaptors.push_back(pDB);

		pDBThread = NULL;
		pDB = NULL;
	}
	
	m_wThreadNum = wThreadNum;
	m_bThreadAutoSelect = bThreadAutoSelect;
	
	//RT_INFO_TRACE("CDBThreadManager::Initialize(), create " << wThreadNum << " DB Threads");
	return rv;	
}

RtResult CDBThreadManager::PostEvent(IDBEvent* pDBEvent, WORD wThreadNo)
{
	RT_ASSERTE(pDBEvent);
	RT_ASSERTE(m_wThreadNum > 0);

	RtResult rv = RT_OK;
	CRtThread* pDBThread = NULL;
	//CDBOciAdaptor* pDB = NULL;
	HANDLE pDB = NULL;
	
	if(wThreadNo > m_wThreadNum)//Out of bound
		return RT_ERROR_FAILURE;

	pDBThread = m_vDBThreads[wThreadNo - 1];
	pDB = m_vDBAdaptors[wThreadNo - 1];
	
	CDBReqEvent* pDBReqEvent = new CDBReqEvent(pDBEvent, m_pCurrEventQueue, pDB);
	rv = pDBThread->GetEventQueue()->PostEvent(pDBReqEvent);
	
	return rv;
}

RtResult CDBThreadManager::PostEvent(IDBEvent* pDBEvent)
{
	RT_ASSERTE(pDBEvent);
	RT_ASSERTE(m_wThreadNum > 0);

	// Case : m_wThreadNum == 1
	if(m_wThreadNum == 1)
		return PostEvent(pDBEvent, 1);

	// Case : m_wThreadNum > 1

	RtResult ret = PostEvent(pDBEvent, m_wCurrentIndex + 1);
	if(RT_SUCCEEDED(ret))
	{
		m_nPushNumber++;
#define MAX_PUSH_NUMBER	10
		if(m_nPushNumber > MAX_PUSH_NUMBER)
		{
			m_wCurrentIndex = 0;
			DWORD dwMinEvents = m_vDBThreads[0]->GetEventQueue()->GetPendingEventsCount();
			DWORD dwTmp = 0;
			
			for(WORD wI=1; wI<m_wThreadNum; wI++)
			{
				dwTmp = m_vDBThreads[wI]->GetEventQueue()->GetPendingEventsCount();
				if(dwMinEvents > dwTmp)
				{
					dwMinEvents = dwTmp;
					m_wCurrentIndex = wI;
				}
			}
			m_nPushNumber = 0;
		}
	}
	else
	{
		RT_WARNING_TRACE("CDBThreadManager::PostEvent, post event error, ret="<<ret);
	}
	return ret;
	/*WORD wMinIndex = 0;
	DWORD dwMinEvents = m_vDBThreads[0]->GetEventQueue()->GetPendingEventsCount();
	RT_INFO_TRACE("CDBThreadManager::PostEvent, m_vDBThreads[0] Num of Events = " << dwMinEvents);
	DWORD dwTmp = 0;
	
	for(WORD wI=1; wI<m_wThreadNum; wI++)
	{
		dwTmp = m_vDBThreads[wI]->GetEventQueue()->GetPendingEventsCount();
		RT_INFO_TRACE("CDBThreadManager::PostEvent, m_vDBThreads[" << wI << "] Num of Events = " << dwTmp);
		if(dwMinEvents > dwTmp)
		{
			dwMinEvents = dwTmp;
			wMinIndex = wI;
		}
	}

	return PostEvent(pDBEvent, wMinIndex + 1);*/
}

void CDBThreadManager::Clean()
{
	RT_INFO_TRACE("CDBThreadManager::Clean()");
	
	WORD wI = 0;
	WORD wSize = m_vDBAdaptors.size();	
	if (DB_TYPE_MYSQL == m_wDBType)
	{
		for(wI = 0; wI < wSize; wI++)
		{
			CDBOciAdaptor* pAdaptor = (CDBOciAdaptor*)m_vDBAdaptors[wI];
			pAdaptor->CloseDatabase();
			delete pAdaptor;
		}
	}
	m_vDBAdaptors.clear();

	wSize = m_vDBThreads.size();	
	for(wI = 0; wI < wSize; wI++)
	{
		m_vDBThreads[wI]->Stop();
		m_vDBThreads[wI]->Join();
		m_vDBThreads[wI]->Destory(0);
	}
	m_vDBThreads.clear();

	m_wThreadNum = 0;
}

WORD CDBThreadManager::GetThreadNum()
{
	return m_wThreadNum;
}
///////////////////////////////////////////
//class CDBReqEvent
///////////////////////////////////////////
CDBReqEvent::CDBReqEvent(
						 IDBEvent* pDBEvent, 
						 IRtEventQueue* pBackEventQueue,
						 HANDLE hDB) 
{ 
	RT_ASSERTE(pDBEvent);
	m_pDBEvent = pDBEvent; 
	m_pBackEventQueue = pBackEventQueue;
	m_hDB = hDB;
}

CDBReqEvent::~CDBReqEvent()
{
}

RtResult CDBReqEvent::OnEventFire()
{
	m_pDBEvent->SetDB(m_hDB);

	//CCmTimeValue tvBegin = CCmTimeValue::GetTimeOfDay();
		
	RtResult rv = m_pDBEvent->OnRequest();
	
	if(m_pDBEvent->NeedResponse())
	{
		CDBRespEvent* pRespEvent = new CDBRespEvent(m_pDBEvent.ParaIn());
		m_pDBEvent = NULL; // Make sure m_pDBEvent->m_pRawPtr be released in the request thread. 
		m_pBackEventQueue->PostEvent(pRespEvent);
	}
	
	//CCmTimeValue tvEnd = CCmTimeValue::GetTimeOfDay();

	//RT_INFO_TRACE("CDBReqEvent::OnEventFire(), time = " << (tvEnd - tvBegin).GetTotalInMsec() << " ms");

	return rv;
}


///////////////////////////////////////////
//class CDBRespEvent
///////////////////////////////////////////
CDBRespEvent::CDBRespEvent(IDBEvent* pDBEvent)
{ 
	RT_ASSERTE(pDBEvent);
	m_pDBEvent = pDBEvent; 
}

CDBRespEvent::~CDBRespEvent()
{
}

RtResult CDBRespEvent::OnEventFire()
{
	RT_ASSERTE(m_pDBEvent->NeedResponse());
	RtResult rv = m_pDBEvent->OnResponse();

	return rv;
}

